/*****************************************************************************
 * server.c: CPM Server application
 *****************************************************************************
 * Copyright (C) 2012
 *
 * Authors: Dan Larkin <danielhlarkin@gmail.com>
 *          Yonatan Naamad <cached@gmail.com>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *****************************************************************************/

#include "datatypes.h"
#include "dict.h"
#include "heap.h"
#include "opcodes.h"
#include <stddef.h>
#include <string.h>
#include <stdio.h>
#include <pthread.h>
#include <time.h>
#include <unistd.h>
#include <errno.h>
#include <arpa/inet.h>

#define CPM_PORT_DATABASE   7589

#define CPM_QUEUE_LENGTH 1000
#define CPM_MAINTENANCE_INTERVAL 5

#define CPM_EXIT_IMPROPER_USAGE 1
#define CPM_EXIT_DATABASE_DOWN 2
#define CPM_EXIT_SOCKET_INIT 3
#define CPM_EXIT_BIND 4
#define CPM_EXIT_NO_TORRENTS 5
#define CPM_EXIT_ALLOC_FAIL 6

#define CPM_MAX( a, b ) ( ( a >= b ) ? a : b )
#define CPM_MIN( a, b ) ( ( a <= b ) ? a : b )

static void cpm_connect_to_database();
static void cpm_get_torrent_list();
static void cpm_setup_listening_socket();
static void cpm_accept_clients();
static void* cpm_maintenance_loop( void *empty );
static int cpm_send_to_client( int client_socket, uint8_t *buffer, int len );
static int cpm_recv_from_client( int client_socket, uint8_t *buffer, int len );
static void cpm_send_to_database( uint8_t *buffer, int len );
static void cpm_recv_from_database( uint8_t *buffer, int len );
static void* cpm_deal_with_client( void *client );
static void cpm_remove_client( cpm_client_node *client );
static int cpm_process_join( cpm_client_node *client, uint8_t *buffer );
static int cpm_process_request( cpm_client_node *client, uint8_t *buffer );
static int cpm_process_free( cpm_client_node *client, uint8_t *buffer );
static void cpm_prune_peer( cpm_torrent_node *torrent,
    cpm_client_node *client );
static void cpm_prune_allocation( cpm_client_node *client,
    cpm_torrent_node *torrent );
static void cpm_make_allocation( cpm_torrent_node *torrent, int32_t bw_min,
    int32_t bw_extra );
static void cpm_record_peer( cpm_torrent_node *torrent,
    cpm_client_node *client );
static void cpm_record_allocation( cpm_client_node *client,
    cpm_torrent_node *torrent, int32_t min, int32_t max );
static int cpm_heap_cmp_clients( void *a, void *b );
static int cpm_heap_cmp_torrents( void *a, void *b );
static void cpm_calculate_average_timestamp( cpm_torrent_node *torrent );
static void cpm_reclaim_resources();
static void cpm_ensure_minimum_allocations( cpm_dict_node *node );
static int cpm_send_alloc( cpm_client_node *client, cpm_torrent_node *torrent,
    int32_t bw_min, int32_t bw_max );
static int cpm_client_useful( cpm_client_node *client, int32_t size );
static cpm_allocation* cpm_search_allocation( cpm_client_node *client,
    cpm_torrent_node *torrent );
static void cpm_remove_small_allocations();
static void cpm_shrink_allocations();

static const uint8_t cpm_server_key[16] = { 0x6e, 0x47, 0x59, 0x78, 0x64, 0x43,
    0x54, 0x77, 0x56, 0x76, 0x6e, 0x4d, 0x36, 0x67, 0x57, 0x72 };

cpm_server_data server_data;

int main( int argc, char **argv )
{
    if( argc != 3 )
        exit( CPM_EXIT_IMPROPER_USAGE );

    pthread_mutex_init( &(server_data.lock), NULL );
    pthread_mutex_lock( &(server_data.lock) );
    
    memcpy( server_data.key, cpm_server_key, 16 );
    server_data.server_port = atoi( argv[1] );
    server_data.database_ip = inet_addr( argv[2] );
    server_data.torrent_dict = cpm_dict_init( 20 );
    server_data.client_dict = cpm_dict_init( 4 );
    server_data.client_heap = cpm_heap_init( cpm_heap_cmp_clients );
    server_data.torrent_heap = cpm_heap_init( cpm_heap_cmp_torrents );
    server_data.temp = calloc( 1, sizeof( cpm_stack ) );
    
    printf("Connecting to Database...\n");
    cpm_connect_to_database();
    printf("Fetching torrent list...\n");
    cpm_get_torrent_list();
    printf("Setting up listening socket...\n");
    cpm_setup_listening_socket();

    pthread_mutex_unlock( &(server_data.lock) );

    printf("Launching maintenance thread...\n");
    while( pthread_create( &(server_data.maintenance_thread), NULL,
        cpm_maintenance_loop, NULL ) );
    
    printf("Begin accepting clients...\n");
    cpm_accept_clients();
    
    return 0;
}

static void cpm_connect_to_database()
{
    printf("\tSetup socket...\n");
    server_data.database_socket = socket( AF_INET, SOCK_STREAM, 0 );
    if ( server_data.database_socket == -1 )
        exit( CPM_EXIT_SOCKET_INIT );
    printf("\t\t%d\n",server_data.database_socket );

    struct sockaddr_in address;
    address.sin_family = AF_INET;
    address.sin_port = htons( CPM_PORT_DATABASE );
    address.sin_addr.s_addr = server_data.database_ip;

    printf("\tConnect...\n");
    int connect_check = connect( server_data.database_socket,
            (struct sockaddr*) &address, sizeof( struct sockaddr_in ) );
    printf("\t\t%d\n",errno);
    if( connect_check != 0 )
        exit( CPM_EXIT_DATABASE_DOWN );
}

static void cpm_get_torrent_list()
{
    cpm_op_exposit response;
    cpm_op_list request;
    request.code = CPM_CODE_LIST;
    memcpy( request.key, cpm_server_key, 16 );
    
    printf("Sending message to database...\n");
    cpm_send_to_database( (uint8_t*) &request, sizeof( cpm_op_list ) );
    printf("Waiting on response from database...\n");
    cpm_recv_from_database( (uint8_t*) &response,
        offsetof( cpm_op_exposit, torrent_list ) );

    printf("Got a response...\n");

    response.len = ntohl( response.len );
    if( response.len == 0 )
        exit( CPM_EXIT_NO_TORRENTS );
        
    printf("Get the %d torrents...\n",response.len);

    response.torrent_list = calloc( response.len, sizeof( cpm_torrent ) );
    if( response.torrent_list == NULL )
        exit( CPM_EXIT_ALLOC_FAIL );

    printf("Getting full torrent info...\n");
    cpm_recv_from_database( (uint8_t*) response.torrent_list,
        response.len * sizeof( cpm_torrent ) );

    printf("Processing rest of torrent info...\n");
    int i;
    cpm_torrent_node *node;
    for( i = 0; i < response.len; i++ )
    {
        node = calloc( 1, sizeof( cpm_torrent_node ) );
        if( node == NULL )
            exit( CPM_EXIT_ALLOC_FAIL );
            
        memcpy( &(node->data.id), &(response.torrent_list[i].id), 20 );
        node->data.min = ntohl( response.torrent_list[i].min );
        node->data.max = ntohl( response.torrent_list[i].max );
        node->data.size = ntohl( response.torrent_list[i].size );
        cpm_dict_insert( server_data.torrent_dict, node );
        node->heap_node = cpm_heap_insert( server_data.torrent_heap, node );
    }
    
    free( response.torrent_list );
}

static void cpm_setup_listening_socket()
{
    server_data.server_socket = socket( AF_INET, SOCK_STREAM, 0 );
    if ( server_data.server_socket == -1 )
        exit( CPM_EXIT_SOCKET_INIT );

    server_data.server_addr.sin_family = AF_INET;
    server_data.server_addr.sin_addr.s_addr = INADDR_ANY;
    server_data.server_addr.sin_port = htons( server_data.server_port );

    if( bind( server_data.server_socket, (struct sockaddr*)
            &server_data.server_addr, sizeof( server_data.server_addr ) ) )
        exit( CPM_EXIT_BIND );

    listen( server_data.server_socket, CPM_QUEUE_LENGTH );
}

static void cpm_accept_clients()
{
    int client_connection;
    cpm_client_node *new_client;

    printf("Begin looping...\n");
    // loop until program termination
    while( 1 )
    {
        printf("Allocate new client.\n");
        new_client = calloc( 1, sizeof( cpm_client_node ) );
        new_client->addrlen = sizeof( struct sockaddr_in );
        
        printf("Accept block...\n");
        client_connection = accept( server_data.server_socket,
            &(new_client->addr), &(new_client->addrlen) );
        printf("Accepted.\n");
        if( client_connection < 0 )
            continue;
            
        new_client->connection = client_connection;
        printf("Spawn thread.\n");
        while( pthread_create( &(new_client->thread), NULL,
            cpm_deal_with_client, new_client ) );
    }
}

static void* cpm_deal_with_client( void *input )
{
    cpm_client_node *client = (cpm_client_node*) input;
    uint8_t *buffer = malloc( 1 ); // make realloc play nice
    uint8_t *temp;
    int status, len, offset;

    while( 1 )
    {
        printf("Wait for opcode.\n");
        temp = realloc( buffer, 1 );
        if( temp == NULL )
        {
            status = -2;
            break;
        }
        else
            buffer = temp;
        status = cpm_recv_from_client( client->connection, buffer, 1 );
        if( status < 0 )
        {
            status = -3;
            break;
        }

        switch( buffer[0] )
        {
            case CPM_CODE_JOIN:
                printf("Got a JOIN.\n");
                len = offsetof( cpm_op_join, torrent_list ) - 1;
                break;
            case CPM_CODE_REQUEST:
                printf("Got a REQUEST.\n");
                len = sizeof( cpm_op_request ) - 1;
                break;
            case CPM_CODE_FREE:
                printf("Got a FREE.\n");
                len = sizeof( cpm_op_free ) - 1;
                break;
            default:
                continue;
        }

        printf("Receiving message body...\n");
        temp = realloc( buffer, len + 1 );
        if( temp == NULL )
        {
            status = -4;
            break;
        }
        else
            buffer = temp;
        
        status = cpm_recv_from_client( client->connection, buffer + 1, len );
        if( status < 0 )
        {
            status = -5;
            break;
        }

        switch( buffer[0] )
        {
            case CPM_CODE_JOIN:
                offset = len + 1;
                len = ntohs( ((cpm_op_join*) buffer)->slots_used ) *
                    sizeof( cpm_torrent );
                    
                temp = realloc( buffer, len + offset );
                if( temp == NULL )
                {
                    status = -6;
                    break;
                }
                else
                    buffer = temp;
                printf("Receiving message body part 2...\n");
                status = cpm_recv_from_client( client->connection,
                    buffer + offset, len );
                if( status < 0 )
                {
                    status = -7;
                    break;
                }
                printf("Processing...\n");
                pthread_mutex_lock( &(server_data.lock) );
                status = cpm_process_join( client, buffer );
                pthread_mutex_unlock( &(server_data.lock) );
                break;
            case CPM_CODE_REQUEST:
                printf("Processing...\n");
                pthread_mutex_lock( &(server_data.lock) );
                status = cpm_process_request( client, buffer );
                pthread_mutex_unlock( &(server_data.lock) );
                break;
            case CPM_CODE_FREE:
                printf("Processing...\n");
                pthread_mutex_lock( &(server_data.lock) );
                status = cpm_process_free( client, buffer );
                pthread_mutex_unlock( &(server_data.lock) );
                break;
            default:
                continue;
        }

        if( status < 0 )
            break;
    }

    if( buffer != NULL )
        free( buffer );
    pthread_mutex_lock( &(server_data.lock) );
    cpm_remove_client( client );
    pthread_mutex_unlock( &(server_data.lock) );
    
    return NULL;
}

static int cpm_send_to_client( int client_socket, uint8_t *buffer, int len )
{
    int chunk;
    int bytes = 0;
    while ( bytes < len )
    {
        chunk = send( client_socket, buffer + bytes, len, 0);
        if( chunk < 0 )
            return chunk;
        bytes += chunk;
    }

    return 0;
}

static int cpm_recv_from_client( int client_socket, uint8_t *buffer, int len )
{
    int chunk;
    int bytes = 0;
    while ( bytes < len )
    {
        chunk = recv( client_socket, buffer + bytes, len, 0);
        if( chunk < 0 )
            return chunk;
        bytes += chunk;
    }

    return 0;
}

static void cpm_send_to_database( uint8_t *buffer, int len )
{
    int chunk;
    int bytes = 0;
    while ( bytes < len )
    {
        chunk = send( server_data.database_socket, buffer + bytes, len, 0);
        if( chunk < 0 )
            exit( CPM_EXIT_DATABASE_DOWN );
        bytes += chunk;
    }
}

static void cpm_recv_from_database( uint8_t *buffer, int len )
{
    int chunk;
    int bytes = 0;
    while ( bytes < len )
    {
        chunk = recv( server_data.database_socket, buffer + bytes, len, 0);
        if( chunk < 0 )
            exit( CPM_EXIT_DATABASE_DOWN );
        bytes += chunk;
    }
}

static void cpm_remove_client( cpm_client_node *client )
{
    cpm_dict_node *node = cpm_dict_lookup( server_data.client_dict, client );
    if( node == NULL )
        return;

    while( client->torrents != NULL )
    {
        cpm_prune_peer( client->torrents->torrent, client );
        cpm_prune_allocation( client, client->torrents->torrent );
    }

    cpm_heap_delete( server_data.client_heap, client->heap_node );
    cpm_dict_delete( server_data.client_dict, node );
}

static void cpm_prune_peer( cpm_torrent_node *torrent, cpm_client_node *client )
{
    cpm_peer *peer = torrent->clients;
    while( peer != NULL )
    {
        if( peer->client == client )
        {
            if( torrent->clients == peer )
                torrent->clients = peer->next;
            if( peer->prev != NULL )
                peer->prev->next = peer->next;
            if( peer->next != NULL )
                peer->next->prev = peer->prev;

            free( peer );
            
            return;
        }
        peer = peer->next;
    }
}

static void cpm_prune_allocation( cpm_client_node *client,
    cpm_torrent_node *torrent )
{
    cpm_allocation *alloc = client->torrents;
    while( alloc != NULL )
    {
        if( alloc->torrent == torrent )
        {
            if( client->torrents == alloc )
                client->torrents = alloc->next;
            if( alloc->prev != NULL )
                alloc->prev->next = alloc->next;
            if( alloc->next != NULL )
                alloc->next->prev = alloc->prev;

            client->bw_used -= alloc->max;
            server_data.bw_used -= alloc->max;
            cpm_heap_decrease_key( server_data.client_heap, client->heap_node );
            torrent->current -= alloc->min;
            
            free( alloc );

            cpm_calculate_average_timestamp( torrent );
            cpm_heap_decrease_key( server_data.torrent_heap,
                torrent->heap_node );
            
            return;
        }
        alloc = alloc->next;
    }
}

static void cpm_record_peer( cpm_torrent_node *torrent,
    cpm_client_node *client )
{
    cpm_peer *current = torrent->clients;
    cpm_peer *new_peer;
    
    while( current != NULL )
    {
        if( current->client == client )
        {
            current->timestamp = time( NULL );
            return;
        }
            
        current = current->next;
    }

    new_peer = calloc( 1, sizeof( cpm_peer ) );
    new_peer->client = client;
    new_peer->prev = current;
    new_peer->timestamp = time( NULL );
    if( torrent->clients == NULL )
        torrent->clients = new_peer;
}

static void cpm_record_allocation( cpm_client_node *client,
    cpm_torrent_node *torrent, int32_t min, int32_t max )
{
    cpm_allocation *new_alloc;
    new_alloc = cpm_search_allocation( client, torrent );
    if( new_alloc == NULL )
    {
        new_alloc = calloc( 1, sizeof( cpm_allocation ) );
        new_alloc->torrent = torrent;
        new_alloc->next = client->torrents;
        if( client->torrents != NULL )
            client->torrents->prev = new_alloc;
        client->torrents = new_alloc;
        client->disk_used += torrent->data.size;
        client->slots_used++;
        server_data.slots_used++;
    }
        
    client->bw_used += max - new_alloc->max;
    server_data.bw_used += max - new_alloc->max;
    torrent->current += min - new_alloc->min;

    if( min - new_alloc->min > 0 )
        client->heap_node = cpm_heap_increase_key( server_data.client_heap,
            client->heap_node );
    else
        cpm_heap_decrease_key( server_data.client_heap, client->heap_node );

    new_alloc->min = min;
    new_alloc->max = max;

    cpm_calculate_average_timestamp( torrent );
    torrent->heap_node = cpm_heap_increase_key( server_data.torrent_heap,
        torrent->heap_node );
}

static int cpm_process_join( cpm_client_node *client, uint8_t *buffer )
{
    cpm_dict_node *node = cpm_dict_lookup( server_data.client_dict, client );
    if( node != NULL )
        return -1;

    cpm_op_join *join = (cpm_op_join*) buffer;
    if( memcmp( join->key, server_data.key, 16 ) )
        return -1;

    client->bw_total = ntohl( join->bw_total );
    server_data.bw_total += client->bw_total;
    client->disk_total = ntohl( join->disk_total );
    client->slots_total = ntohs( join->slots_total );    

    client->heap_node = cpm_heap_insert( server_data.client_heap, client );

    // check and add listed torrents
    int16_t i;
    cpm_torrent *torrents = (cpm_torrent*) ( buffer + offsetof( cpm_op_join,
        torrent_list ) );
    for( i = 0; i < ntohs( join->slots_used ); i++ )
    {
        node = cpm_dict_lookup( server_data.torrent_dict, torrents[i].id );
        if( node == NULL )
            continue;
            
        torrents[i].min = ntohl( torrents[i].min );
        torrents[i].max = ntohl( torrents[i].max );

        cpm_record_peer( node->element, client );
        cpm_record_allocation( client, node->element, torrents[i].min,
            torrents[i].max );
    }

    cpm_dict_insert( server_data.client_dict, client );

    return 0;
}

static int cpm_process_request( cpm_client_node *client, uint8_t *buffer )
{
    cpm_dict_node *node = cpm_dict_lookup( server_data.client_dict, client );
    if( node == NULL )
        return -1;

    cpm_op_request *request = (cpm_op_request*) buffer;
    node = cpm_dict_lookup( server_data.torrent_dict, request->torrent_id );
    if( node == NULL )
        return -1;

    cpm_make_allocation( node->element, 0, ntohl( request->bw ) );

    return 0;
}

static int cpm_process_free( cpm_client_node *client, uint8_t *buffer )
{
    cpm_dict_node *node = cpm_dict_lookup( server_data.client_dict, client );
    if( node == NULL )
        return -1;

    cpm_op_free *free = (cpm_op_free*) buffer;
    cpm_dict_node *torrent_node = (cpm_dict_node*)
        cpm_dict_lookup( server_data.torrent_dict, free->torrent_id );
    cpm_allocation * alloc = cpm_search_allocation( client,
        torrent_node->element );
    cpm_record_allocation( client, torrent_node->element, alloc->min,
        alloc->max - ntohl( free->bw ) );

    cpm_heap_decrease_key( server_data.client_heap, client->heap_node );

    return 0;
}

static int cpm_send_alloc( cpm_client_node *client, cpm_torrent_node *torrent,
    int32_t bw_min, int32_t bw_max )
{
    cpm_op_allocate alloc;
    alloc.code = CPM_CODE_ALLOCATE;
    alloc.bw_min = htonl( bw_min );
    alloc.bw_max = htonl( bw_max );
    memcpy( alloc.torrent_id, torrent->data.id, 20 );

    int status = cpm_send_to_client( client->connection, (uint8_t*) &alloc,
        sizeof( cpm_op_allocate ) );

    return status;
}

/**
 * Compares the fractional free bandwidth of two clients, returning a non-zero
 * value only if the first client has strictly more free bandwidth than the
 * second.
 *
 * @param a First client
 * @param b Second client
 * @return  Non-zero if first client has more, otherwise zero
 */
static int cpm_heap_cmp_clients( void *a, void *b )
{
    cpm_client_node *client_a = (cpm_client_node*) a;
    cpm_client_node *client_b = (cpm_client_node*) b;
    
    int64_t a_used = ((int64_t) client_a->bw_used) << 32;
    int64_t b_used = ((int64_t) client_b->bw_used) << 32;
    int64_t a_total = ((int64_t) client_a->bw_total);
    int64_t b_total = ((int64_t) client_b->bw_total);

    return ( ( a_used / a_total ) < ( b_used / b_total ) );
}

/**
 * Compares the average timestamp of the torrents allocations.
 *
 * @param a First torrent
 * @param b Second torrent
 * @return  Non-zero if first torrent's average is older, otherwise zero
 */
static int cpm_heap_cmp_torrents( void *a, void *b )
{
    cpm_torrent_node *torrent_a = (cpm_torrent_node*) a;
    cpm_torrent_node *torrent_b = (cpm_torrent_node*) b;
    
    return ( torrent_a->average_time < torrent_b->average_time );
}

static void cpm_calculate_average_timestamp( cpm_torrent_node *torrent )
{
    int32_t n = 0;
    int64_t sum = 0;
    cpm_peer *peer = torrent->clients;

    while( peer != NULL )
    {
        n++;
        sum += peer->timestamp;
        peer = peer->next;
    }
    
    n = CPM_MAX( n, 1 );

    torrent->average_time = sum / n;
}

static void* cpm_maintenance_loop( void *empty )
{
    while( 1 )
    {
        printf("Running Maintenance...\n");
        pthread_mutex_lock( &(server_data.lock) );
        
        if( server_data.bw_used > 0.9 * server_data.bw_total )
            cpm_reclaim_resources();
            
        cpm_ensure_minimum_allocations( server_data.torrent_dict->root);

        pthread_mutex_unlock( &(server_data.lock) );
        
        sleep( CPM_MAINTENANCE_INTERVAL );
    }

    return NULL;
}

static void cpm_reclaim_resources()
{
    cpm_remove_small_allocations();
    if( server_data.bw_used > 0.85 * server_data.bw_total )
        cpm_shrink_allocations();
}

static void cpm_remove_small_allocations()
{
    cpm_torrent_node *torrent;
    cpm_client_node *client;
    cpm_peer *peer, *next;
    cpm_allocation *alloc;

    while( server_data.bw_used > 0.85 * server_data.bw_total &&
        !cpm_heap_empty( server_data.torrent_heap ) )
    {
        torrent = cpm_heap_find_min( server_data.torrent_heap );

        if( torrent->current > torrent->data.min )
        {
            peer = torrent->clients;
            while( peer != NULL )
            {
                next = peer->next;
                client = peer->client;
                alloc = cpm_search_allocation( client, torrent );
                if( alloc->min < torrent->current - torrent->data.min )
                {
                    cpm_send_alloc( client, torrent, 0, 0 );
                    cpm_prune_allocation( client, torrent );
                    cpm_prune_peer( torrent, client );
                }

                peer = next;
            }
        }

        torrent = cpm_heap_delete_min( server_data.torrent_heap );
        cpm_stack_push( server_data.temp, torrent );
    }

    while( !cpm_stack_empty( server_data.temp ) )
    {
        torrent = (cpm_torrent_node*) cpm_stack_pop( server_data.temp );
        torrent->heap_node = cpm_heap_insert( server_data.torrent_heap,
            torrent );
    }
}

static void cpm_shrink_allocations()
{
    cpm_torrent_node *torrent;
    cpm_client_node *client;
    cpm_peer *peer;
    cpm_allocation *alloc;
    int32_t alloc_min, alloc_max, current_min, current_max;

    while( server_data.bw_used > 0.85 * server_data.bw_total &&
        !cpm_heap_empty( server_data.torrent_heap ) )
    {
        torrent = cpm_heap_find_min( server_data.torrent_heap );

        if( torrent->current > torrent->data.min )
        {
            peer = torrent->clients;
            client = peer->client;
            alloc = cpm_search_allocation( client, torrent );

            current_min = alloc->min;
            current_max = alloc->max;
            alloc_min = current_min - ( torrent->current - torrent->data.min );
            alloc_max = current_max - ( torrent->current - torrent->data.max );
                
            cpm_send_alloc( client, torrent, alloc_min, alloc_max );
            cpm_record_allocation( client, torrent, alloc_min, alloc_max );
            cpm_record_peer( torrent, client );
        }

        torrent = cpm_heap_delete_min( server_data.torrent_heap );
        cpm_stack_push( server_data.temp, torrent );
    }

    while( !cpm_stack_empty( server_data.temp ) )
    {
        torrent = (cpm_torrent_node*) cpm_stack_pop( server_data.temp );
        torrent->heap_node = cpm_heap_insert( server_data.torrent_heap,
            torrent );
    }
}

static void cpm_ensure_minimum_allocations( cpm_dict_node *node )
{
    if( node == NULL )
        return;

    cpm_torrent_node *torrent = (cpm_torrent_node*) node->element;
    if( torrent->current < torrent->data.min )
    {
        cpm_make_allocation( torrent, torrent->data.min - torrent->current,
            -1 );
    }

    cpm_ensure_minimum_allocations( node->left );
    cpm_ensure_minimum_allocations( node->right );
}

static void cpm_make_allocation( cpm_torrent_node *torrent, int32_t bw_min,
    int32_t bw_extra )
{
    int32_t alloc_min, alloc_max, current_min, current_max;
    int32_t min = bw_min;
    int32_t max = ( bw_extra == -1 ) ? min : min + bw_extra;
 
    int status;
    
    printf("Making allocation...\n");
    cpm_client_node *client;
    cpm_peer *peer = torrent->clients;
    cpm_allocation *allocation;
    while( max > 0 && !cpm_heap_empty( server_data.client_heap ) )
    {
        printf("\tFinding client...\n");
        while( peer != NULL && !cpm_client_useful( peer->client,
                torrent->data.size ) )
            peer = peer->next;
            
        if( peer == NULL )
        {
            client = (cpm_client_node*)
                cpm_heap_find_min( server_data.client_heap );
        }
        else
            client = peer->client;

        if( cpm_client_useful( client, torrent->data.size ) )
        {
            printf("\tFound useful client...\n");
            allocation = cpm_search_allocation( client, torrent );
            current_min = ( allocation != NULL ) ? allocation->min : 0;
            current_max = ( allocation != NULL ) ? allocation->max : 0;
            printf("Currents: %d,%d\n",current_min,current_max);
            
            alloc_min = CPM_MAX( CPM_MIN( client->bw_total - client->bw_used,
                min ) + current_min, 1 );
            alloc_max = ( bw_extra == -1 ) ? CPM_MAX( alloc_min, current_max ) :
                CPM_MIN( client->bw_total - client->bw_used, max ) +
                current_max;

            printf("\tSending allocation request...\n");
            status = cpm_send_alloc( client, torrent, alloc_min, alloc_max );
            if( status == 0 )
            {
                printf("\tAllocation request sent! (%d,%d) of (%d,%d)\n", alloc_min, alloc_max, min, max);
                cpm_record_peer( torrent, client );
                cpm_record_allocation( client, torrent, alloc_min, alloc_max );
                min -= alloc_min;
                max -= alloc_max;
            }
            else
                printf("\tAllocation request failed!\n");
        }
        else
        {
            client = (cpm_client_node*)
                cpm_heap_delete_min( server_data.client_heap );
            cpm_stack_push( server_data.temp, client );            
        }
    }
    
    if( ( min == bw_min ) && ( max == ( ( bw_extra == -1 ) ? min : min + bw_extra ) ) )
        printf("\tNobody useful!\n");
    
    while( !cpm_stack_empty( server_data.temp ) )
    {
        client = (cpm_client_node*) cpm_stack_pop( server_data.temp );
        client->heap_node = cpm_heap_insert( server_data.client_heap, client );
    }
}

static int cpm_client_useful( cpm_client_node *client, int32_t size )
{
    int available_bw = client->bw_total > client->bw_used;
    int available_slots = client->slots_total > client->slots_used;
    int available_disk = client->disk_total >= client->disk_used + size;

    return( available_bw && available_slots && available_disk );
}

static cpm_allocation* cpm_search_allocation( cpm_client_node *client,
    cpm_torrent_node *torrent )
{
    cpm_allocation *alloc = client->torrents;
    while( alloc != NULL )
    {
        if( alloc->torrent == torrent )
            return alloc;

        alloc = alloc->next;
    }

    return NULL;
}
